package net.bwie.ads;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class Ads_car_out {
    public static void main(String[] args) {
        // 1.表执行环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.输入表 读取Kafka数据
        createInputTable(tableEnv);

        // 3.数据处理
        Table table = handle(tableEnv);

//        tableEnv.createTemporaryView("resultTable", table);
//        TableResult tableResult = tableEnv.executeSql("select * from resultTable");
//        tableResult.print();

        // 4.映射表 写入doris表中
        createOutputTable(tableEnv);

        // 5.保存数据
        saveToSink(tableEnv, table);
    }

    private static void saveToSink(TableEnvironment tableEnv, Table table) {
        // todo 创建临时视图 然后写入doris中
        tableEnv.createTemporaryView("dwstable", table);
        //INSERT INTO dws_summary_cars_log
        tableEnv.executeSql(
                "insert into ads_car_out " +
                        "select * from dwstable");
    }

    private static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }

    private static void createInputTable(TableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE dws_summary_cars_log (\n" +
                // 窗口开始时间
                "window_start TIMESTAMP(3),\n" +
                // 窗口结束时间
                "window_end TIMESTAMP(3),\n" +
                // 时间
                "time_ltz TIMESTAMP(3),\n" +
                // 小区ID
                "cId INT,\n" +
                // 驶入次数
                "in_cnt INT,\n" +
                // 驶出次数
                "out_cnt INT\n" +
                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                "'fenodes' = 'node102:8030',\n" +
                "'table.identifier' = 'dws_sca.dws_summary_cars_log',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456',\n" +
                "'sink.label-prefix' = 'dws_summary_cars_id123')\n"
        );

    }


    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(
                        " SELECT  DATE_FORMAT(window_start, '%Y-%m-%d') as time,\n" +
                                " '驶入' as name, round(in_cnt/total*100,2) as zb \n" +
                        " FROM ( \n" +
                        "     SELECT \n" +
                        "         DATE_FORMAT(window_start,'%Y-%m-%d') as time, \n" +
                        "         SUM(in_cnt) as in_cnt, \n" +
                        "         SUM(out_cnt) as out_cnt, \n" +
                        "         SUM(in_cnt+out_cnt) as total \n" +
                        "     FROM dws_sca.dws_summary_cars_log \n" +
                        "     GROUP BY DATE_FORMAT(window_start,'%Y-%m-%d') \n" +
                        " ) a1 \n" +
                        " UNION ALL \n" +
                        " SELECT DATE_FORMAT(window_start, '%Y-%m-%d') as time,\n" +
                                " '驶出' as name, round(out_cnt/total*100,2) as zb \n" +
                        " FROM ( \n" +
                        "     SELECT \n" +
                        "         DATE_FORMAT(window_start,'%Y-%m-%d') as time, \n" +
                        "         SUM(in_cnt) as in_cnt, \n" +
                        "         SUM(out_cnt) as out_cnt, \n" +
                        "         SUM(in_cnt+out_cnt) as total \n" +
                        "     FROM dws_sca.dws_summary_cars_log \n" +
                        "     GROUP BY DATE_FORMAT(window_start,'%Y-%m-%d') \n" +
                        " ) a2 "
        );
        return table;
    }

    private static void createOutputTable(TableEnvironment tableEnv) {
      tableEnv.executeSql("CREATE TABLE ads_cars_out_cnt (\n" +
                "time STRING,\n" +
                "name STRING,\n" +
                "zb DOUBLE\n" +
                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                "'fenodes' = 'node102:8030',\n" +
                "'table.identifier' = 'ads_sca.ads_cars_out_cnt',\n"
        );

    }
}
